package com.shujia.sql

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo2DSLWordCount {
  def main(args: Array[String]): Unit = {
    /**
     * 在新版本的spark中，如果想要编写spark sql的话，需要使用新的spark入口类：SparkSession
     */
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local")
      .appName("wc spark sql")
      .getOrCreate()

    /**
     * spark sql和spark core的核心数据类型不太一样
     *
     * 1、读取数据构建一个DataFrame,相当于一张表
     */
    val linesDF: DataFrame = sparkSession.read
      .format("csv") //指定读取数据的格式
      .schema("line STRING") //指定列的名和列的类型，多个列之间使用,分割
      .option("sep", "\n") //指定分割符，csv格式读取默认是英文逗号
      .load("spark/data/words.txt") // 指定要读取数据的位置，可以使用相对路径

    /**
     * DSL: 类SQL语法 api  介于代码和纯sql之间的一种api
     *
     * spark在DSL语法api中，将纯sql中的函数都使用了隐式转换变成一个scala中的函数
     * 如果想要在DSL语法中使用这些函数，需要导入隐式转换
     *
     */
    //导入Spark sql中所有的sql隐式转换函数
    import org.apache.spark.sql.functions._
    //导入另一个隐式转换，后面可以直接使用$函数引用字段进行处理
    import sparkSession.implicits._

//    linesDF.select(explode(split($"line","\\|")) as "word")
//      .groupBy($"word")
//      .count().show()

    val resultDF: DataFrame = linesDF.select(explode(split($"line", "\\|")) as "word")
      .groupBy($"word")
      .agg(count($"word") as "counts")

    /**
     * 保存数据
     */
    resultDF
      .repartition(1)
      .write
      .format("csv")
      .option("sep","\t")
      .mode(SaveMode.Overwrite)
      .save("spark/data/sqlout2")

  }

}
